package com.atguigu.flink.chapter07;

import com.atguigu.flink.chapter05.Source.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/1/20 14:06
 */
public class Flink17_Watermark_Summary {
    public static void main(String[] args) throws Exception {
        /**
         *  TODO 一、watermark概念
         *          1、衡量 事件时间 的进展
         *          2、是一个 特殊的时间戳， 生成之后，随着 流的流动 而 向后传递
         *          3、用来处理 数据乱序 的问题
         *          4、触发 窗口等 的 计算、关闭
         *          5、单调递增的 （时间不能倒退）
         *          6、Flink认为，小于watermark时间戳的 数据 处理完了，不应该再出现
         *
         *  TODO 二、watermark的生成方式
         *          1、周期性：发射watermark定义在 onPeriodicEmit()方法里，默认200ms
         *          2、间歇性：发射watermark定义在 onEvent()方法里，来一条数据执行一次
         *
         *  TODO 三、watermark代码指定
         *          官方提供两个写法，都是基于周期性的：
         *          assignTimestampsAndWatermarks(
         *                         WatermarkStrategy
         *                                 .<T>forMonotonousTimestamps()  // 升序
         *                                 .withTimestampAssigner()     // 指定 ET
         *                                 )
         *          assignTimestampsAndWatermarks(
         *                         WatermarkStrategy
         *                                 .<T>forBoundedOutOfOrderness(Duration)   // 乱序
         *                                 .withTimestampAssigner()     // 指定 ET
         *                                 )
         *
         *  TODO 四、watermark多并行度下的传递
         *          1、以最小的为准， 类比木桶原理
         *          2、一对多 => 广播
         *             多对一 => 以最小的为准
         *             多对多 => 上面两个的结合
         *
         *  TODO 五、处理迟到数据
         *          1、窗口允许迟到（关窗之前的迟到数据）
         *              1） 当 wm >= 窗口最大时间戳时， 会 触发，但是不会关窗
         *              2） 当 窗口最大时间戳 + 允许迟到时间 >= wm >= 窗口最大时间戳, 每来一条迟到数据，都会进入窗口，触发
         *              3） 当 wm >= 窗口最大时间戳 + 允许迟到时间 ， 会关窗， 迟到数据不再进入窗口，不会再触发
         *          2、侧输出流（关窗之后的迟到数据）
         *              1） outputtag，要用匿名内部类的写法： new OutputTag<泛型>("名字"){}
         *              2） 从主流里面获取
         *
         *
         *
         *  flink处理乱序、迟到：
         *      1、watermark设置 乱序程度（等待时间）
         *      2、窗口允许迟到
         *      3、侧输出流
         *
         *
         *
         *
         *
         *
         *
         *
         *
         *
         *
         */

    }

}
